#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include "report.h"
#include "cJSON.h"

//#define ADDRESS     "tcp://192.168.50.64:1883"//服务器的地址以及端口号
#define CLIENTID    "code-mqtt"
#define TOPIC_DATA_D    "/app/data/down"//订阅的主题
#define TOPIC_DATA_U    "/app/data/up"
#define TOPIC_CTRL_D    "/app/control/down"
#define TOPIC_CTRL_U    "/app/control/up"
#define TOPIC_MOD_D    "/app/mode/down"
#define TOPIC_MOD_U    "/app/mode/up"
#define QOS         1
#define TIMEOUT     10000L
#define N     6 //M0设备的个数
#define M     5 //modbus设备的个数
#define P     6 //mapp设备的个数

//定义的全局变量
MQTTClient client;
MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
MQTTClient_message pubmsg = MQTTClient_message_initializer;
MQTTClient_deliveryToken token;
int rc;

volatile MQTTClient_deliveryToken deliveredtoken;

struct report reportData;
struct mqtt_server mqttData;
struct memory share_memory;
static struct shm_param para_m0;
static struct shm_param para_modbus;
static struct shm_param para_mapp;
struct msgbuf msg_send,msg_rcv;

//读取点表信息并填充结构提
void readTable()
{   
    //打开点表
    FILE *fb = NULL;
    fb = fopen("./data_list.json","r+");
    if (fb == NULL)
    {
        printf("data_list opren error\n");
        return;
    }

    //获得文件大小
    struct stat statbuf;
    stat("./data_list.json",&statbuf);
    int fileSize = statbuf.st_size;
    printf("文件大小:%d\n",fileSize);

    char *jsonBuf = (char *)malloc(sizeof(char) * fileSize + 1);
    memset(jsonBuf,0,fileSize + 1);

    int size = fread(jsonBuf,sizeof(char),fileSize,fb);
    if (size == 0)
    {
        printf("读取文件失败\n");
        return;
    }
    
    fclose(fb);

    //获得json的变量
    cJSON *root = cJSON_Parse(jsonBuf);
    if (!root)
    {
        const char *err = cJSON_GetErrorPtr();
        printf("Error before : [%s]\n",err);
        free((void*)err);
        free(jsonBuf);
        return;
    }
    free(jsonBuf);

    cJSON *item = NULL;

    //解析report的数据
    item = cJSON_GetObjectItem(root,"report");
    if (item != NULL)
    {
        cJSON *val = NULL;

        val = cJSON_GetObjectItem(item,"type");
        if (val != NULL && val->type == cJSON_Number)
        {
            reportData.type = val->valueint;
            printf("type :%d\n",reportData.type);
        }
        
        val = cJSON_GetObjectItem(item,"period");
        if (val != NULL && val->type == cJSON_Number)
        {
            reportData.period = val->valueint;
            printf("period :%d\n",reportData.period);
        }

        free(val);
        val = NULL;
    }

    //解析mqtt_server的数据
    item = cJSON_GetObjectItem(root,"mqtt_server");
    if (item != NULL)
    {
        cJSON *val = NULL;

        val = cJSON_GetObjectItem(item,"addr");
        if (val != NULL && val->type == cJSON_String)
        {
            mqttData.addr = val->valuestring;
            printf("addr:%s\n",mqttData.addr);
        }

        val = cJSON_GetObjectItem(item,"port");
        if (val != NULL && val->type == cJSON_Number)
        {
            mqttData.port = val->valueint;
            printf("port :%d\n",mqttData.port);
        }
        free(val);
        val = NULL;
    }
    
    free(root);
    free(item);
}

//MQTT的其中一个回调函数
void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}
//MQTT的其中一个回调函数完

void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

//周期上报的函数
void period_report(void)
{
    int i;
    cJSON *arry1[N];
    cJSON *arry2[M];
    cJSON *arry3[P];
    char value_tmp[10]={0};
    cJSON *up = cJSON_CreateObject();
    cJSON_AddItemToObject(up,"type",cJSON_CreateNumber(1));
    cJSON_AddItemToObject(up,"result",cJSON_CreateNumber(0));
    cJSON *data = cJSON_CreateArray();
    cJSON_AddItemToObject(up,"data",data);
    for(i=0;i<N;i++)
    {
        arry1[i] = cJSON_CreateObject();
        cJSON_AddItemToArray(data,arry1[i]);
        cJSON_AddItemToObject(arry1[i],"key",cJSON_CreateNumber(share_memory.memory_m0[i].key));
        if (share_memory.memory_m0[i].type == 1 || share_memory.memory_m0[i].type == 2)
        {
            memset(value_tmp,0,sizeof(value_tmp));
            sprintf(value_tmp,"%d",share_memory.memory_m0[i].new_val.i_val);
            cJSON_AddItemToObject(arry1[i],"val",cJSON_CreateString(value_tmp));
        }
        else
        {
            memset(value_tmp,0,sizeof(value_tmp));
            sprintf(value_tmp,"%f",share_memory.memory_m0[i].new_val.f_val);
            cJSON_AddItemToObject(arry1[i],"val",cJSON_CreateString(value_tmp));
        }
    }

    for ( i = 0; i < M; i++)
    {
        arry2[i] = cJSON_CreateObject();
        cJSON_AddItemToArray(data,arry2[i]);
        cJSON_AddItemToObject(arry2[i],"key",cJSON_CreateNumber(share_memory.memory_modbus[i].key));
        if (share_memory.memory_modbus[i].type == 1 || share_memory.memory_modbus[i].type == 2)
        {
            memset(value_tmp,0,sizeof(value_tmp));
            sprintf(value_tmp,"%d",share_memory.memory_modbus[i].new_val.i_val);
            cJSON_AddItemToObject(arry2[i],"val",cJSON_CreateString(value_tmp));
        }
        else
        {
            memset(value_tmp,0,sizeof(value_tmp));
            sprintf(value_tmp,"%f",share_memory.memory_modbus[i].new_val.f_val);
            cJSON_AddItemToObject(arry2[i],"val",cJSON_CreateString(value_tmp));
        }
    }

    for ( i = 0; i < P; i++)
    {
        arry3[i] = cJSON_CreateObject();
        cJSON_AddItemToArray(data,arry3[i]);
        cJSON_AddItemToObject(arry3[i],"key",cJSON_CreateNumber(share_memory.memory_mapp[i].key));
        if (share_memory.memory_mapp[i].type == 1 || share_memory.memory_mapp[i].type == 2)
        {
            memset(value_tmp,0,sizeof(value_tmp));
            sprintf(value_tmp,"%d",share_memory.memory_mapp[i].new_val.i_val);
            cJSON_AddItemToObject(arry3[i],"val",cJSON_CreateString(value_tmp));
        }
        else
        {
            memset(value_tmp,0,sizeof(value_tmp));
            sprintf(value_tmp,"%f",share_memory.memory_mapp[i].new_val.f_val);
            cJSON_AddItemToObject(arry3[i],"val",cJSON_CreateString(value_tmp));
        }
    }
    
    char buf[1024] = {0};
    strcpy(buf,cJSON_Print(up));
    pubmsg.payload = buf;
    pubmsg.payloadlen = (int)strlen(buf);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    MQTTClient_publishMessage(client,TOPIC_DATA_U,&pubmsg,&token);
}

void change_report()
{
    int i;
    char value_tmp[10] = {0};
    cJSON *arry1[N];
    cJSON *arry2[M];
    cJSON *arry3[P];
    cJSON *up = cJSON_CreateObject();
    cJSON_AddItemToObject(up,"type",cJSON_CreateNumber(1));
    cJSON_AddItemToObject(up,"result",cJSON_CreateNumber(0));
    cJSON *data = cJSON_CreateArray();
    cJSON_AddItemToObject(up,"data",data);

    for ( i = 0; i < N; i++)
    {
        if (share_memory.memory_m0[i].new_val.i_val != share_memory.memory_m0[i].old_val.i_val)
        {
            arry1[i] = cJSON_CreateObject();
            cJSON_AddItemToArray(data,arry1[i]);
            memset(value_tmp,0,sizeof(value_tmp));
            cJSON_AddItemToObject(arry1[i],"key",cJSON_CreateNumber(share_memory.memory_m0[i].key));
            if (share_memory.memory_m0[i].type == 1 || share_memory.memory_m0[i].type == 2)
            {
                sprintf(value_tmp,"%d",share_memory.memory_m0[i].new_val.i_val);
                cJSON_AddItemToObject(arry1[i],"val",cJSON_CreateString(value_tmp));
            }
            else
            {
                sprintf(value_tmp,"%f",share_memory.memory_m0[i].new_val.f_val);
                cJSON_AddItemToObject(arry1[i],"val",cJSON_CreateString(value_tmp));
            } 
        }
    }

    for ( i = 0; i < M; i++)
    {
        if (share_memory.memory_modbus[i].new_val.i_val != share_memory.memory_modbus[i].old_val.i_val)
        {
            arry2[i] = cJSON_CreateObject();
            cJSON_AddItemToArray(data,arry2[i]);
            memset(value_tmp,0,sizeof(value_tmp));
            cJSON_AddItemToObject(arry2[i],"key",cJSON_CreateNumber(share_memory.memory_modbus[i].key));
            if (share_memory.memory_modbus[i].type == 1 || share_memory.memory_modbus[i].type == 2)
            {
                sprintf(value_tmp,"%d",share_memory.memory_modbus[i].new_val.i_val);
                cJSON_AddItemToObject(arry2[i],"val",cJSON_CreateString(value_tmp));
            }
            else
            {
                sprintf(value_tmp,"%f",share_memory.memory_modbus[i].new_val.f_val);
                cJSON_AddItemToObject(arry2[i],"val",cJSON_CreateString(value_tmp));
            }
        }
    }
    
    for ( i = 0; i < P; i++)
    {
        if (share_memory.memory_mapp[i].new_val.i_val != share_memory.memory_mapp[i].old_val.i_val)
        {
            arry3[i] = cJSON_CreateObject();
            cJSON_AddItemToArray(data,arry3[i]);
            memset(value_tmp,0,sizeof(value_tmp));
            cJSON_AddItemToObject(arry3[i],"key",cJSON_CreateNumber(share_memory.memory_mapp[i].key));
            if (share_memory.memory_mapp[i].type == 1 || share_memory.memory_mapp[i].type == 2)
            {
                sprintf(value_tmp,"%d",share_memory.memory_mapp[i].new_val.i_val);
                cJSON_AddItemToObject(arry3[i],"val",cJSON_CreateString(value_tmp));
            }
            else
            {
                sprintf(value_tmp,"%f",share_memory.memory_mapp[i].new_val.f_val);
                cJSON_AddItemToObject(arry3[i],"val",cJSON_CreateString(value_tmp));
            }
        }
    }
    
    char buf[1024] = {0};
    strcpy(buf,cJSON_Print(up));
    pubmsg.payload = buf;
    pubmsg.payloadlen = (int)strlen(buf);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    MQTTClient_publishMessage(client,TOPIC_DATA_U,&pubmsg,&token);

}

int moode_change(char *message)
{
    int m_type,m_period = 5;
    cJSON *root = cJSON_Parse(message);
    cJSON *item = cJSON_GetObjectItem(root,"data");
    if (item == NULL)
    {
        printf("get object error\n");
        return -1;
    }
    cJSON *vall = cJSON_GetObjectItem(item,"type");
    if (vall == NULL)
    {
        printf("get object type error\n");
        return -1;
    }
    m_type = vall->valueint;
    if (m_type == 2)
    {
        vall = cJSON_GetObjectItem(item,"period");
        if (vall == NULL)
        {
            printf("get object period error\n");
            return -1;
        }
        m_period = vall->valueint;
    }
        
    root = NULL;
    item = NULL;

   //打开点表
    FILE *fb = NULL;
    fb = fopen("./data_list.json","r");
    if (fb == NULL)
    {
        printf("data_list opren error\n");
        return -1;
    }

    //获得文件大小
    struct stat statbuf;
    stat("./data_list.json",&statbuf);
    int fileSize = statbuf.st_size;
    printf("文件大小:%d\n",fileSize);

    char *jsonBuf = (char *)malloc(sizeof(char) * fileSize + 1);
    memset(jsonBuf,0,fileSize + 1);

    int size = fread(jsonBuf,sizeof(char),fileSize,fb);
    if (size == 0)
    {
        printf("读取文件失败\n");
        return -1;
    } 
    
    root = cJSON_Parse(jsonBuf);
    if (!root)
    {
        const char *err = cJSON_GetErrorPtr();
        printf("Error before : [%s]\n",err);
        free((void*)err);
        free(jsonBuf);
        return -1;
    }

    item = cJSON_GetObjectItem(root,"report");
    cJSON_ReplaceItemInObject(item,"type",cJSON_CreateNumber(m_type));
    cJSON_ReplaceItemInObject(item,"period",cJSON_CreateNumber(m_period));

    memset(jsonBuf,0,fileSize + 1);
    jsonBuf = cJSON_Print(root);
    printf("修改后的点表:%s\n",jsonBuf);
    FILE *fb_w = NULL;
    fb_w = fopen("./data_list.json","w");
    if (fb_w == NULL)
    {
        perror("open file error\n");
        return -1;
    }
    
    size_t writes = fwrite(jsonBuf,sizeof(char),fileSize,fb_w);
    if(writes == 0)
    {
        printf("修改失败\n");
        return -1;
    }

    fclose(fb);
    fclose(fb_w);

    //回复给客户端
    cJSON *rett;
    rett = cJSON_CreateObject();
    cJSON_AddNumberToObject(rett,"result",0);
    cJSON_AddStringToObject(rett,"msg","success!");
    char msg_ret[64];
    strcpy(msg_ret,cJSON_Print(rett));

    pubmsg.payload = msg_ret;
    pubmsg.payloadlen = (int)strlen(msg_ret);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    printf("上报模式修改回复测试\n");
    MQTTClient_publishMessage(client, TOPIC_MOD_U , &pubmsg, &token);

    exit(0);
    return 0;
}

int ctrl_change(char *message)
{
    int type;
    int mkey;
    char *val = NULL;
    char buff[128] = {0};
    cJSON *item = NULL;
    cJSON *root = cJSON_Parse(message);
    printf("%s\n",cJSON_Print(root));
    item = cJSON_GetObjectItem(root,"type");
    if (item != NULL)
    {
        type = item->valueint;
        printf("type:%d\n",type);
    }

    item = cJSON_GetObjectItem(root,"data");
    if (item == NULL)
    {
        printf("get data error\n");
        return -1;
    }
    cJSON *value = NULL;
    value = cJSON_GetObjectItem(item,"key");
    mkey = value->valueint;
    printf("key:%d\n",mkey);

    value = cJSON_GetObjectItem(item,"val");
    val = value->valuestring;
    printf("val:%s\n",val);

    memset(msg_send.mdata,0,sizeof(msg_send.mdata));
    memset(msg_rcv.mdata,0,sizeof(msg_rcv.mdata));
    strcpy(msg_send.mdata,message);

    if (mkey >0 && mkey<7)
    {
        printf("发送给m0\n");
        //m0设备
        msg_send.mtype = 1;
        if (msg_queue_send("send_m0",&msg_send,sizeof(msg_send),0) < 0)
        {
            printf("msg_queue_send to M0 error\n");
            return -1;
        }

        if (msg_queue_recv("recv_m0",&msg_rcv,sizeof(msg_rcv),0,0) < 0)
        {
            printf("msg_queue_recv error\n");
        }
        printf("接收到来自M0的消息%s\n",msg_rcv.mdata);
    }
    else if (mkey>100 && mkey<106)
    {
        printf("发送给modbus\n");
        //modbus设备
        msg_send.mtype = 2;
        if (msg_queue_send("send_modbus",&msg_send,sizeof(msg_send),0) < 0)
        {
            printf("msg_queue_send to modbus error\n");
        }
        if (msg_queue_recv("recv_modbus",&msg_rcv,sizeof(msg_rcv),0,0) < 0)
        {
            printf("msg_queue_recv error\n");
        }
        printf("接收到来自modbud的消息%s\n",msg_rcv.mdata);
    }
    else if (mkey > 200 && mkey < 207)
    {
        printf("发送给mapp\n");
        //mapp设备
        msg_send.mtype = 3;
        if (msg_queue_send("send_mapp",&msg_send,sizeof(msg_send),0) < 0)
        {
            printf("msg_queue_send to mapp error\n");
        }
        if (msg_queue_recv("recv_mapp",&msg_rcv,sizeof(msg_rcv),0,0) < 0)
        {
            printf("msg_queue_recv error\n");
        }
        printf("接收到来自mapp的消息%s\n",msg_rcv.mdata);
    }
    strcpy(buff,msg_rcv.mdata);
    pubmsg.payload = buff;
    pubmsg.payloadlen = (int)strlen(buff);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    MQTTClient_publishMessage(client,TOPIC_CTRL_U,&pubmsg,&token);
    free(root);
    root = NULL;
    free(item);
    item = NULL;

    return 0;
}

//主动发送数据的线程
void *func_send(void *arg)
{
    switch (reportData.type)
    {
    case 0:
        printf("不上报\n");
        while (1);
        break;
    case 1:
        printf("变化上报\n");
        while (1)
        {
            change_report();
            sleep(reportData.period);
        }
        break;
    case 2:
        printf("周期上报\n");
        while (1)
        {
            period_report();
            sleep(reportData.period);
        }
        break;
    }
    return NULL;
}

//MQTT接收函数
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    printf("Message arrived\n");
    printf("topic: %s\n", topicName);
    printf("message: ");

    if (strcmp(topicName,TOPIC_DATA_D) == 0)
    {
        printf("刷新上报\n");
        period_report();
    }
    else if (strcmp(topicName,TOPIC_MOD_D) == 0)
    {
        printf("上报模式的修改\n");
        moode_change(message->payload);
    }
    else if (strcmp(topicName,TOPIC_CTRL_D) == 0)
    {
        printf("控制指令\n");
        ctrl_change(message->payload);
    }
    
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}
//MQTT接收函数完

int main(int argc, char const *argv[])
{
    int ret_memory = -1;//用于共享内存创建成功与否的判断
    pthread_t tid_send;//用于发送的线程号

    //解析点表获得数据
    readTable();

    char address[25]={0};
    sprintf(address,"tcp://%s:%d",mqttData.addr,mqttData.port);

    //创建了一个mqtt客户端
    MQTTClient_create(&client,address, CLIENTID,
        MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;//链接选项
    conn_opts.cleansession = 1;
    int ch;

    //设置回调函数
    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);
    
    //链接broker
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    //订阅主题
    MQTTClient_subscribe(client, TOPIC_DATA_D, QOS);
    MQTTClient_subscribe(client, TOPIC_CTRL_D, QOS);
    MQTTClient_subscribe(client, TOPIC_MOD_D, QOS);

    //开辟M0设备的共享内存
    ret_memory = shm_init(&para_m0,"shm_m0",sizeof(struct std_node)*N);
    if(ret_memory < 0)
    {
        printf("shm_m0 init error.\n");
        return -1;
    }
    share_memory.memory_m0 = shm_getaddr(&para_m0);
    if (share_memory.memory_m0 == NULL)
    {
        printf("shm_m0 error\n");
        return -1;
    }
    ret_memory = -1;
    
    //开辟modbus的共享内存
    ret_memory = shm_init(&para_modbus,"shm_modbus",sizeof(struct std_node)*M);
    if(ret_memory < 0)
    {
        printf("shm_modbus init error.\n");
        return -1;
    }
    share_memory.memory_modbus = shm_getaddr(&para_modbus);
    if (share_memory.memory_modbus == NULL)
    {
        printf("shm_modbus error\n");
        return -1;
    }
    ret_memory = -1;

    //开辟mapp的共享内存
    ret_memory = shm_init(&para_mapp,"shm_mapp",sizeof(struct std_node)*P);
    if(ret_memory < 0)
    {
        printf("shm_mapp init error.\n");
        return -1;
    }
    share_memory.memory_mapp = shm_getaddr(&para_mapp);
    if(share_memory.memory_mapp == NULL)
    {
        printf("shm_mapp error\n");
        return -1;
    }

    //开辟线程向客户端发送内容6
    if (pthread_create(&tid_send,NULL,func_send,NULL))
    {
        perror("create thread_send error");
        return -1;
    }
    

    //保持阻塞防止程序提前结束
    do 
    {
        ch = getchar();
    } while(ch!='Q' && ch != 'q');

    MQTTClient_unsubscribe(client, TOPIC_CTRL_D);
    MQTTClient_unsubscribe(client, TOPIC_MOD_D);
    MQTTClient_unsubscribe(client, TOPIC_DATA_D);
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);

    
    return 0;
}
